package com.kotlin.lib.flow

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlin.time.ExperimentalTime
import kotlin.time.measureTime

fun requestFlow(int: Int) = flow<String> {
    emit("first request $int")
    delay(100)
    emit("second request $int")

}

@ExperimentalTime
fun main() = runBlocking<Unit> {

    val flowOfFlow = (1..3).asFlow().onEach { delay(150) }
    val t = System.currentTimeMillis()

    flowOfFlow.map {
        if (it == 2) throw IllegalAccessException("$it exception")
        "map $it"
    }.onStart {
        println("start")
    }.onCompletion {
        println("done ${it?.cause}")
    }.catch {
        println(it.message)
    }.collect {
        println(it)
    }
//    flatMapLatest(flowOfFlow, t)
}

private suspend fun flatMapLatest(flowOfFlow: Flow<Int>, t: Long) {
    flowOfFlow.flatMapLatest {
        requestFlow(it)
    }
        .collect {
            println("$it @${System.currentTimeMillis() - t}")
        }
}

private suspend fun flatMapMerge(flowOfFlow: Flow<Int>, t: Long) {
    flowOfFlow.flatMapMerge {
        requestFlow(it)
    }.collect {
        println("$it @${System.currentTimeMillis() - t}")
    }
}

private suspend fun flatMapConcat(flowOfFlow: Flow<Int>, t: Long) {
    flowOfFlow//flatMapConcat:它们在等待内部流完成之前开始收集下⼀个值
        .flatMapConcat {
            requestFlow(it)
        }.collect {
            println("$it @${System.currentTimeMillis() - t}")
        }
}

private suspend fun combine() {
    val intFlow = arrayOf(1, 2, 3).asFlow().onEach { delay(100) }
    val stringFlow = arrayOf("one", "two", "three", "four").asFlow().onEach { 150 }

    intFlow.combine(stringFlow) { int, string ->
        "$int-$string"
    }.collect {
        println(it)
    }
}

private suspend fun zip() {
    val intFlow = arrayOf(1, 2, 3).asFlow()
    val stringFlow = arrayOf("one", "two", "three", "four").asFlow()

    intFlow.zip(stringFlow) { int, string ->
        "$int-$string"
    }.collect {
        println(it)
    }
}

@ExperimentalTime
private suspend fun collectLastest() {
    val time = measureTime {
        foo().collectLatest {
            delay(300)
            println(it)
        }
    }
    println(time)
}

private suspend fun conflateTest() {
    foo()
        .conflate()// 合并发射项，不对每个值进⾏处理.流发送方比较快，处理方比较慢，处理方只处理最近的一个，忽略中间的值
        .collect {
            delay(300)// 假装我们花费 300 毫秒来处理它
            println(it)
        }
}